# ----------------------------------------------------------------------
# ' split mbr to daily units
# ----------------------------------------------------------------------

load_library = c('bit64','data.table','fst','future.apply','stringr','logger','vroom')
invisible(lapply(load_library, function(x) library(x, character.only=TRUE, quietly= TRUE)))

library(parallel)
library(tsibble)

bucket = file.path('/N','project','iuni_doctorshopping')

split_sample_to_week = function(yw, outfile=NULL){

	mbr_yw_sample = mbr_detail[start_week <= yw & yw <= end_week, ]

	if (is.null(outfile)) {
		return(mbr_yw_sample)
	} else {
		write_fst(mbr_yw_sample, outfile, 100)
		return(NULL)
	}
}

	universe = 'ses'

	infile = file.path(bucket,'raw_fst_v4_202101',universe,paste0(universe,'_mbr_enroll.fst'))
	outfolder = file.path(bucket,'derived_v4_202101','mbr_split','week',universe)
	if (!dir.exists(outfolder)){
		dir.create(outfolder, recursive = TRUE)	
	}

	mbr_detail = read_fst(infile, as.data.table = TRUE)
	
	mbr_detail[, start_week := tsibble::yearweek(as.Date(ELIGEFF))]
	mbr_detail[, end_week := tsibble::yearweek(as.Date(ELIGEND))]

	list_week = unique(tsibble::yearweek(seq.Date(from=as.Date('2019-01-01'),to=as.Date('2020-09-30'),by=1)))
	ff_list_week = gsub(' ','',as.character(list_week))

	lits_week_outfile = file.path(outfolder,paste0(universe,'_mbr_enroll_',ff_list_week,'.fst'))
	
	lapply(1:length(list_week), function(i) {		

		if (file.exists(lits_week_outfile[i])){

			logger::log_info('already completed for ',lits_week_outfile[i])

		} else {

			logger::log_info('now running for ',lits_week_outfile[i])
			split_sample_to_week(list_week[i], lits_week_outfile[i])

		}
	})



